package topics.impl;

import java.util.concurrent.ThreadPoolExecutor;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import pubsub.Subscriber;
import pubsub.Subscription;

import topics.Topic;
import topics.TopicListener;

public class SubscriptionTopicListener implements TopicListener {

	private static Log LOG = LogFactory
			.getLog(SubscriptionTopicListener.class.getName());

	private Subscription m_subscription;

	/**
	 * Construct a listener instance.
	 * 
	 * @param subscription
	 *            the subscription which is being wrapped in a listener
	 */
	public SubscriptionTopicListener(Subscription subscription) {
		m_subscription = subscription;
	}

	/**
	 * @return Subscription
	 */
	public Subscription getSubscription() {
		return m_subscription;
	}

	/**
	 * DOCUMENT_ME
	 * 
	 * @param topic
	 *            DOCUMENT_ME
	 */
	public void topicChanged(Topic topic) {
		if (m_subscription != null) {
			try {
				notify(topic.getCurrentMessage());
			} catch (Exception e) {
				LOG.debug("Notification for topic " + topic
						+ " failed for subscription: " + m_subscription, e);
			}
		}
	}

	private void notify(Object rawMsg) throws Exception {
		LOG.debug("Notification being sent for subscription with id "
				+ m_subscription.getID() + "; message value: " + rawMsg);

		Subscriber consumer = m_subscription.getSubscriber();

		consumer.notify(rawMsg);
	}
}
